package com.swak.reactivex.transport.quic;

import static java.util.concurrent.TimeUnit.MILLISECONDS;

import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.io.InputStream;
import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.util.Collection;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.swak.Constants;
import com.swak.exception.NoSupportException;
import com.swak.exception.RemotingException;
import com.swak.exception.ServerException;
import com.swak.reactivex.transport.ChannelHandler;
import com.swak.reactivex.transport.Codec;
import com.swak.reactivex.transport.Endpoint;
import com.swak.reactivex.transport.NettyCodec;
import com.swak.reactivex.transport.NettyServerHandler;
import com.swak.reactivex.transport.Server;
import com.swak.reactivex.transport.codec.MultipleCodec;
import com.swak.reactivex.transport.resources.LoopResources;
import com.swak.registry.URL;
import com.swak.utils.StringUtils;

import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.PooledByteBufAllocator;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.channel.ChannelOption;
import io.netty.channel.ChannelPipeline;
import io.netty.handler.ssl.NettySslKit;
import io.netty.handler.timeout.IdleStateHandler;
import io.netty.incubator.codec.quic.InsecureQuicTokenHandler;
import io.netty.incubator.codec.quic.QuicServerCodecBuilder;
import io.netty.incubator.codec.quic.QuicSslContext;
import io.netty.incubator.codec.quic.QuicSslContextBuilder;
import io.netty.incubator.codec.quic.QuicStreamChannel;

public class QuicNettyServer extends Endpoint implements Server {

	protected Logger logger = LoggerFactory.getLogger(QuicNettyServer.class);

	private Channel channel;
	private Bootstrap bootstrap;
	private InetSocketAddress localAddress;
	private InetSocketAddress bindAddress;

	/**
	 * the cache for alive worker channel. <ip:port, telnet channel>
	 */
	private final Map<String, com.swak.reactivex.transport.Channel> channels;
	private final LoopResources resources;

	public QuicNettyServer(URL url, LoopResources resources, ChannelHandler handler) throws RemotingException {
		super(url, handler);
		this.localAddress = getUrl().toInetSocketAddress();
		String bindIp = getUrl().getParameter(Constants.Bind_Host);
		if (StringUtils.isBlank(bindIp)) {
			bindIp = getUrl().getHost();
		}
		this.bindAddress = new InetSocketAddress(bindIp, getUrl().getPort());
		this.channels = new ConcurrentHashMap<>();
		this.resources = resources;

		try {
			this.start();
			if (logger.isInfoEnabled()) {
				logger.info(
						"Start " + getClass().getSimpleName() + " bind " + bindAddress + ", export " + localAddress);
			}
		} catch (Throwable t) {
			throw new RemotingException("Failed to bind " + getClass().getSimpleName() + " on " + localAddress
					+ ", cause: " + t.getMessage(), t);
		}
	}

	@Override
	public Collection<com.swak.reactivex.transport.Channel> getChannels() {
		return channels.values();
	}

	@Override
	public void start() throws ServerException {
		if (this.started) {
			return;
		}
		this.started = true;
		bootstrap = new Bootstrap();
		bootstrap.option(ChannelOption.SO_REUSEADDR, Boolean.TRUE) //
				.option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT);//

		// 配置： ServerBootstrap
		this.customBootstrap(bootstrap);

		// 配置处理器
		bootstrap.handler(new QuicServerCodecBuilder().sslContext(this.buildQuicSslContext())
				.maxIdleTimeout(5000, TimeUnit.MILLISECONDS)//
				.initialMaxData(10000000)//
				.initialMaxStreamDataBidirectionalLocal(1000000)//
				.initialMaxStreamDataBidirectionalRemote(1000000)//
				.initialMaxStreamsBidirectional(100)//
				.initialMaxStreamsUnidirectional(100)//
				.tokenHandler(InsecureQuicTokenHandler.INSTANCE)//
				// 不用设置处理器
				.handler(new ChannelInboundHandlerAdapter() {
					@Override
					public boolean isSharable() {
						return true;
					}
				})
				// 设置流处理器
				.streamHandler(new QuicChannelInitializer() {
					@Override
					protected void initChannel(QuicStreamChannel ch) throws Exception {
						super.initChannel(ch);
						ChannelPipeline pipeline = ch.pipeline();
						QuicNettyServer.this.customChannel(ch);
						pipeline.addLast("server-idle", new IdleStateHandler(0, 0, getIdleTimeout(), MILLISECONDS));
						pipeline.addLast("handler", new NettyServerHandler(getUrl(), QuicNettyServer.this, channels));
					}
				}).build());
		ChannelFuture channelFuture = bootstrap.bind(this.bindAddress);
		this.channel = channelFuture.syncUninterruptibly().channel();
	}

	private QuicSslContext buildQuicSslContext() {
		try {
			InputStream keyIs = this.loadResource(this.url().getParameter(Constants.SSL_keyPath));
			InputStream certificateIs = this.loadResource(this.url().getParameter(Constants.SSL_certPath));
			return QuicSslContextBuilder.forServer(NettySslKit.toPrivateKey(keyIs, null), null,
					NettySslKit.toX509Certificates(certificateIs)).applicationProtocols("swakx/1.0").build();
		} catch (Exception e) {
			throw new NoSupportException(e);
		}
	}

	private InputStream loadResource(String localtion) throws FileNotFoundException {
		if (StringUtils.startsWith("files:", localtion)) {
			return new FileInputStream(StringUtils.removeStart(localtion, "files:"));
		}
		return QuicNettyServer.class.getResourceAsStream("/" + StringUtils.removeStart(localtion, "classpath:"));
	}

	@Override
	public void close() {
		if (logger.isInfoEnabled()) {
			logger.info("Close " + getClass().getSimpleName() + " bind " + bindAddress);
		}
		try {
			super.close();
		} catch (Throwable e) {
			logger.warn(e.getMessage(), e);
		}
		try {
			doClose();
		} catch (Throwable e) {
			logger.warn(e.getMessage(), e);
		}
	}

	protected void doClose() throws Throwable {
		try {
			if (channel != null) {
				// unbind.
				channel.close();
			}
		} catch (Throwable e) {
			logger.warn(e.getMessage(), e);
		}
		try {
			if (channels != null && channels.size() > 0) {
				for (com.swak.reactivex.transport.Channel channel : channels.values()) {
					try {
						channel.close();
					} catch (Throwable e) {
						logger.warn(e.getMessage(), e);
					}
				}
			}
		} catch (Throwable e) {
			logger.warn(e.getMessage(), e);
		}
		try {
			if (channels != null) {
				channels.clear();
			}
		} catch (Throwable e) {
			logger.warn(e.getMessage(), e);
		}
	}

	@Override
	public void send(Object message, boolean sent) throws RemotingException {
		Collection<com.swak.reactivex.transport.Channel> channels = this.channels.values();
		for (com.swak.reactivex.transport.Channel channel : channels) {
			if (channel.isConnected()) {
				channel.send(message, sent);
			}
		}
	}

	@Override
	public SocketAddress localAddress() {
		return localAddress;
	}

	@Override
	public SocketAddress remoteAddress() {
		return bindAddress;
	}

	@Override
	public void connected(com.swak.reactivex.transport.Channel ch) throws RemotingException {
		// If the server has entered the shutdown process, reject any new connection
		if (this.isClosing() || this.isClosed()) {
			logger.warn("Close new channel " + ch
					+ ", cause: server is closing or has been closed. For example, receive a new connect request while in shutdown process.");
			ch.close();
			return;
		}
		super.connected(ch);
	}

	/**
	 * 配置编解码协议
	 * 
	 * @param ch
	 */
	protected void customChannel(QuicStreamChannel ch) {
		ChannelPipeline pipeline = ch.pipeline();
		Codec codec = new MultipleCodec(getUrl());
		NettyCodec nettyCodec = new NettyCodec(codec, getUrl(), this);
		pipeline.addLast(NettyCodec.Encoder_Name, nettyCodec.getEncoder());
		pipeline.addLast(NettyCodec.Decoder_Name, nettyCodec.getDecoder());
	}

	/**
	 * 配置事件组，配置各种参数
	 * 
	 * @param bootstrap
	 */
	protected void customBootstrap(Bootstrap bootstrap) {
		bootstrap.group(resources.onServer()).channel(resources.onDatagramChannel());
	}
}